
package cn.com.lyb.flink.partition;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * DataSet 批处理 单词统计
 */
public class PartitionDemo {
    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> stringDataStreamSource = env.socketTextStream("172.32.1.6", 1234);
        // 随机分区 random.nextInt(this.numberOfChannels) (下游算子并行度)
        // stringDataStreamSource.shuffle().print();

        // rebalance 轮询  (this.nextChannelToSendTo + 1) % this.numberOfChannels; (下游算子并行度)
        // 如果是 数据源倾斜的场景， source读进来之后，调用rebalance， 来解决数据倾斜
        //stringDataStreamSource.rebalance().print();

        // rescale 缩放  实现轮询，局部组队，比rebalance高效
        // stringDataStreamSource.rescale().print();

        // broadcast 广播  发送所有下游
        // stringDataStreamSource.broadcast().print();

        // global 全局分区 这种做法非常极端，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行度变成了1，所以使用这个操作需要非常谨慎，可能对程序造成很大的压力。
        stringDataStreamSource.global().print();
        // keyby  按照指定的key去发送，相同的key发往同一个子任务

        env.execute();
    }
}
